package pub.cleangao.memm.transport.netty;


import io.netty.bootstrap.ServerBootstrap;
import io.netty.buffer.PooledByteBufAllocator;
import io.netty.channel.*;
import io.netty.channel.nio.NioEventLoopGroup;
import io.netty.channel.socket.SocketChannel;
import io.netty.channel.socket.nio.NioServerSocketChannel;
import io.netty.handler.logging.LogLevel;
import io.netty.handler.logging.LoggingHandler;
import io.netty.handler.timeout.IdleStateHandler;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import pub.cleangao.memm.registry.RpcRuntime;
import pub.cleangao.memm.transport.RpcServer;
import pub.cleangao.memm.transport.netty.codec.MessageDecoder;
import pub.cleangao.memm.transport.netty.codec.MessageEncoder;
import pub.cleangao.memm.util.ThreadPoolFactory;
import java.net.InetAddress;
import java.util.concurrent.TimeUnit;


@Slf4j
public class NettyRpcServer implements RpcServer {

    private EventLoopGroup bossGroup;
    private EventLoopGroup workerGroup;


    @SneakyThrows
    public void start() {
        Runtime.getRuntime().addShutdownHook(new Thread(RpcRuntime::downShelf));
        String host = (RpcRuntime.getConfigProperties().getBindIp() == null || RpcRuntime.getConfigProperties().getBindIp().isBlank())
                ? InetAddress.getLocalHost().getHostAddress() : RpcRuntime.getConfigProperties().getBindIp();
        final int CPU_CORE_COUNT = Runtime.getRuntime().availableProcessors();
        ServerBootstrap serverBootstrap = new ServerBootstrap();
        bossGroup = new NioEventLoopGroup(1, ThreadPoolFactory.createThreadFactory("rpc-server", true));
        workerGroup = new NioEventLoopGroup(CPU_CORE_COUNT * 2, ThreadPoolFactory.createThreadFactory("rpc-server", true));
        try {
            serverBootstrap.group(bossGroup, workerGroup)
                    .channel(NioServerSocketChannel.class)
                    .childOption(ChannelOption.ALLOCATOR, PooledByteBufAllocator.DEFAULT)
                    .childOption(ChannelOption.TCP_NODELAY, true)
                    .childOption(ChannelOption.SO_KEEPALIVE, true)
                    .option(ChannelOption.SO_BACKLOG, 1024 * 1024 * 1024)
                    .handler(new LoggingHandler(LogLevel.INFO))
                    .childHandler(new ChannelInitializer<SocketChannel>() {
                        @Override
                        protected void initChannel(SocketChannel ch) {
                            ChannelPipeline p = ch.pipeline();
                            p.addLast(new IdleStateHandler(30, 0, 0, TimeUnit.SECONDS));
                            p.addLast(new MessageEncoder());
                            p.addLast(new MessageDecoder());
                            p.addLast(new NettyRpcServerHandler());
                        }
                    });
            int port = RpcRuntime.getConfigProperties().getPort();
            log.info("Rpc serve-node(this) is listening on port :{}", port);
            ChannelFuture f = serverBootstrap.bind(host, port).sync();
            f.channel().closeFuture().sync();
        } catch (InterruptedException e) {
            log.error("Exception when start server: ", e);
            e.printStackTrace();
        } finally {
            bossGroup.shutdownGracefully();
            workerGroup.shutdownGracefully();
        }
    }

    @Override
    public void close() {
        bossGroup.shutdownGracefully();
        workerGroup.shutdownGracefully();
    }

}
